package com.wzf.greattruth.zookeeper.election;

import java.util.HashSet;
import java.util.Set;
import java.util.SortedSet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import com.alibaba.fastjson.JSON;
import com.wzf.greattruth.util.DateUtil;
import com.wzf.greattruth.util.HessianUtils;
import com.wzf.greattruth.zookeeper.ZookeeperManager;
import com.wzf.greattruth.zookeeper.helper.ZookeeperHelper;
import com.wzf.greattruth.zookeeper.lock.DistributedLock;
import com.wzf.greattruth.zookeeper.node.LeaderOffer;
import com.wzf.greattruth.zookeeper.node.LeaderOfferData;
import com.wzf.greattruth.zookeeper.node.Node;

/**
 * 选举处理器
 * 
 * @author wuzhengfei
 *
 */
public class ElectionHandler implements Watcher {
    private static final Logger LOGGER = LoggerFactory.getLogger(ElectionHandler.class);
    /**
     * 超过多长时间未发送心跳信息认为节点已死亡
     */
    private static final int DEFAULT_DEAD_TIMEOUT = 1000 * 60;
    private static final int DEFAULT_CHECK_INTERVAL = 1000 * 3;
    private static final String ELECTION_NODE_PREFIX = "ele-";

    private ZooKeeper zooKeeper;
    /**
     * 选举数据在zk中存放的节点位置
     */
    private String dir;
    /**
     * 当前选举人的名字
     */
    private String offerName;

    /**
     * 选举状态
     */
    private ElectionState state = ElectionState.STOP;

    /**
     * 分布式锁数据存放的目录
     */
    private String lockRootDir;
    /**
     * 分布式锁的节点key
     */
    private String lockNodeKey;

    /**
     * 当前的选举对象
     */
    private LeaderOffer leaderOffer;

    private int deadTimeout = DEFAULT_DEAD_TIMEOUT;
    /**
     * 每隔多长时间检查一次
     */
    private int checkInterval = DEFAULT_CHECK_INTERVAL;
    private DistributedLock distributedLock;
    private ZookeeperManager zookeeperManager;
    private Set<ElectionEventAware<?>> electionEventAwares = new HashSet<>();

    private ScheduledExecutorService scheduledExecutorService;

    public ElectionHandler(ZooKeeper zooKeeper, String dir, String offerName, String lockRootDir, String lockNodeKey) {
        this(zooKeeper, dir, offerName, lockRootDir, lockNodeKey, DEFAULT_DEAD_TIMEOUT, DEFAULT_CHECK_INTERVAL);
    }

    public ElectionHandler(ZooKeeper zooKeeper, String dir, String offerName, String lockRootDir, String lockNodeKey,
            int deadTimeout, int checkInterval) {
        super();
        if (zooKeeper == null || StringUtils.isBlank(dir) || StringUtils.isBlank(offerName)
                || StringUtils.isBlank(lockRootDir) || StringUtils.isBlank(lockNodeKey)) {
            throw new IllegalStateException("param error!");
        }

        this.zooKeeper = zooKeeper;
        this.dir = dir;
        this.offerName = offerName;
        this.lockNodeKey = lockNodeKey;
        this.lockRootDir = lockRootDir;
        this.deadTimeout = deadTimeout;
        this.checkInterval = checkInterval;
        init();
    }

    private void init() {
        // 初始状态设置为stop
        distributedLock = new DistributedLock(zooKeeper);
        zookeeperManager = ZookeeperManager.getInstance();

        boolean rootDirExist = zookeeperManager.exist(zooKeeper, dir);
        if (!rootDirExist) {
            boolean creatDirResult = zookeeperManager.createDirIfNotExist(zooKeeper, dir);
            if (!creatDirResult) {
                throw new RuntimeException("创建目录失败，dir=" + dir);
            }
        }

        // 启动定时线程，定时发送心跳，删除已经死亡的节点
        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    // 发送心跳信息
                    sendHeartbeat();

                    // 主动清理失效节点
                    removeDeadNodes();

                    determineState();

                } catch (Exception e) {
                    LOGGER.error("heartbeat or remove dead node failed!", e);
                }
            }
        }, checkInterval, checkInterval, TimeUnit.MILLISECONDS);
    }

    /**
     * 启动,检测当前属于什么节点
     * 
     */
    public boolean start() {
        if (leaderOffer == null) {
            // 在zookeeper上创建一个临时有序节点，并将当前时间、offerName作为数据存储到节点中
            LeaderOfferData offer = new LeaderOfferData();
            offer.setLastHeartbeat(System.currentTimeMillis());
            offer.setOfferName(offerName);
            byte[] datas = HessianUtils.encode(offer);

            String path = ZookeeperHelper.getNodePath(dir, ELECTION_NODE_PREFIX);
            String nodeFullPath = zookeeperManager.createNode(zooKeeper, path, datas, CreateMode.EPHEMERAL_SEQUENTIAL);
            if (StringUtils.isBlank(nodeFullPath)) {
                return false;
            }

            String nodeId = ZookeeperHelper.getNodeId(nodeFullPath);
            // 创建leader offer
            leaderOffer = new LeaderOffer(dir, nodeId, ELECTION_NODE_PREFIX, offerName);
        }

        // 检查当前offer处于什么状态
        determineState();
        return true;
    }

    public boolean stop() {
        if (leaderOffer == null) {
            changeState(ElectionState.STOP);
            return true;
        }
        if (ElectionState.STOP == state) {
            return true;
        }
        String nodePath = ZookeeperHelper.getNodePath(leaderOffer.getDir(), leaderOffer.getId());
        // 节点不存在，则直接认为已停止
        boolean exist = zookeeperManager.exist(zooKeeper, nodePath);
        if (!exist) {
            changeState(ElectionState.STOP);
            return true;
        }
        boolean deleted = zookeeperManager.deleteNode(zooKeeper, nodePath);
        if (!deleted) {
            changeState(ElectionState.ERROR);
            return false;
        } else {
            leaderOffer = null;
            changeState(ElectionState.STOP);
            return true;
        }
    }

    /**
     * 获取Leader信息
     * 
     * @return
     */
    public LeaderOffer getLeader() {
        SortedSet<Node> nodes = zookeeperManager.getChildren(zooKeeper, dir, ELECTION_NODE_PREFIX);
        if (CollectionUtils.isEmpty(nodes)) {
            return null;
        }
        Node node = nodes.first();
        byte[] bytes = zookeeperManager.getDatas(zooKeeper, node);
        LeaderOfferData data = HessianUtils.decode(bytes);
        LeaderOffer leader = new LeaderOffer(node.getDir(), node.getId(), ELECTION_NODE_PREFIX, data.getOfferName());
        return leader;
    }

    public boolean isLeader() {
        return state == ElectionState.LEADER;
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getType() == EventType.NodeDeleted) {
            if (state != ElectionState.STOP) {
                determineState();
            }
        }
    }

    /**
     * 探测当前状态
     * <li>如果
     * 
     * @return
     */
    private synchronized void determineState() {
        SortedSet<Node> nodes = zookeeperManager.getChildren(zooKeeper, dir, ELECTION_NODE_PREFIX);
        if (CollectionUtils.isEmpty(nodes)) {
            changeState(ElectionState.STOP);
            return;
        }
        Node minNode = nodes.first();
        if (minNode.getSequence() == leaderOffer.getSequence()) {
            changeState(ElectionState.LEADER);
        } else {
            SortedSet<Node> preNodes = nodes.headSet(leaderOffer);
            Node preNode = preNodes.last();
            String preNodeId = preNode.getId();
            String preNodePath = ZookeeperHelper.getNodePath(dir, preNodeId);
            boolean exist = zookeeperManager.exist(zooKeeper, preNodePath, this);
            if (exist) {
                changeState(ElectionState.FLLOWER);
            } else {
                determineState();
            }
        }
    }

    /**
     * 修改状态
     * 
     * @param newState:修改后的状态
     */
    private void changeState(ElectionState newState) {
        ElectionState oldState = state;
        if (oldState == null || newState == null) {
            throw new IllegalStateException("非法状态异常");
        }
        if (oldState == newState) {
            return;
        }

        // 鉴别事件
        ElectionEventType type = null;
        switch (newState) {
        case ERROR:
            type = ElectionEventType.ERROR;
            break;
        case FLLOWER:
            type = ElectionEventType.FLLOWING;
            break;
        case LEADER:
            type = ElectionEventType.LEADING;
            break;
        case STOP:
            type = ElectionEventType.STOP;
            break;
        default:
            break;
        }
        if (type == null) {
            throw new IllegalStateException("非法事件类型异常, currentState=" + newState);
        }
        this.state = newState;

        // 停止scheduler
        if (ElectionState.STOP == this.state) {
            scheduledExecutorService.shutdown();
        }

        // aware操作
        synchronized (this) {
            if (!CollectionUtils.isEmpty(electionEventAwares)) {
                for (ElectionEventAware<?> aware : electionEventAwares) {
                    aware.onElectionEvent(type);
                }
            }
        }
    }

    /**
     * 发送心跳信息
     * 
     * @return
     */
    private boolean sendHeartbeat() {
        boolean enable = isEnable();
        if (!enable) {
            return false;
        }
        try {
            String nodePath = ZookeeperHelper.getNodePath(leaderOffer.getDir(), leaderOffer.getId());
            boolean exist = zookeeperManager.exist(zooKeeper, nodePath);
            if (!exist) {
                stop();
                return false;
            }

            // 查询节点的数据
            byte[] datas = zookeeperManager.getDatas(zooKeeper, leaderOffer);
            LeaderOfferData offerData = null;
            if (datas == null) {
                offerData = new LeaderOfferData();
                offerData.setOfferName(leaderOffer.getOfferName());
            } else {
                offerData = HessianUtils.decode(datas);
            }
            offerData.setLastHeartbeat(System.currentTimeMillis());
            byte[] heartbeatDatas = HessianUtils.encode(offerData);

            // 更新节点的数据
            boolean updateResult = zookeeperManager.updateDatas(zooKeeper, leaderOffer, heartbeatDatas);
            return updateResult;
        } catch (Throwable e) {
            LOGGER.error("sendHeartbeat failed1 leaderOffer={}", JSON.toJSONString(leaderOffer), e);
            return false;
        }
    }

    /**
     * 删除失效的节点
     * <li>此方法可以保证分布式的情况下，仅有一个线程执行删除操作
     * <li>单个几点删除失败时，跳过，继续删除下一个节点
     * 
     * @return
     */
    private boolean removeDeadNodes() {
        boolean lock = distributedLock.lock(lockRootDir, lockNodeKey);
        if (!lock) {
            return false;
        }
        try {
            // 查询节点
            SortedSet<Node> nodes = zookeeperManager.getChildren(zooKeeper, dir, ELECTION_NODE_PREFIX);
            if (CollectionUtils.isEmpty(nodes)) {
                return true;
            }
            long current = System.currentTimeMillis();
            for (Node node : nodes) {
                boolean remove = false;
                // 查询节点
                byte[] datas = zookeeperManager.getDatas(zooKeeper, node);
                LeaderOfferData offerData = null;
                if (datas == null) {
                    // 如果节点数据为空，认为是无效节点，直接删除
                    remove = true;
                } else {
                    // 如果当前时间和节点上次心跳时间差超过了限制，那么认为节点已死
                    offerData = HessianUtils.decode(datas);
                    long lastHeartbeat = offerData.getLastHeartbeat();
                    if (current - lastHeartbeat > deadTimeout) {
                        remove = true;
                    }
                }

                // 删除节点
                if (remove) {
                    System.err.println("remove nodeid=" + node.getId() + "  offername=" + offerData.getOfferName()
                            + " beattime=" + DateUtil.formatTime(offerData.getLastHeartbeat()) + "  current="
                            + DateUtil.formatTime(current) + "  delta=" + (current - offerData.getLastHeartbeat()));
                    String nodePath = ZookeeperHelper.getNodePath(node.getDir(), node.getId());
                    boolean removeResult = zookeeperManager.deleteNode(zooKeeper, nodePath);
                    if (!removeResult) {
                        LOGGER.error("deleteNode failed! node={}", JSON.toJSONString(node));
                    }
                }
            }
            return true;

        } catch (Throwable e) {
            LOGGER.error("removeDeadNodes failed!", e);
            return false;
        } finally {
            distributedLock.unlock(lockRootDir, lockNodeKey);
        }
    }

    public void addElectionEventAware(ElectionEventAware<?> aware) {
        electionEventAwares.add(aware);
    }

    private boolean isEnable() {
        if (leaderOffer == null) {
            return false;
        }
        return true;
    }

}
